package com.example.tutorial.java.concurrent.async;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 异步回调
 * https://www.cnblogs.com/fingerboy/p/9948736.html
 *
 * @author: cph
 * @date: 2021-8-12
 */
public class CompletableFutureTest {

    private static ThreadPoolExecutor executor =
            new ThreadPoolExecutor(2, 2, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

    public static void main(String[] args) throws Exception {

        /*===================================================================================/
        /**
         * supplyAsyncy用于创建执行有返回值的任务
         * 如果没有传入Executor对象将会使用ForkJoinPool.commonPool()作为它的线程池执行异步代码
         * 在实际使用中,一般我们使用自己创建的线程池对象来作为参数传入使用
         */
        CompletableFuture<Double> supplyFuture = CompletableFuture.supplyAsync(CompletableFutureTest::fetchPrice);
        CompletableFuture<Double> supplyExecutorFuture = CompletableFuture.supplyAsync(CompletableFutureTest::fetchPrice, executor);
        // 获取返回值方式一：get()会阻塞当前调用线程
        Double r1 = supplyFuture.get();
        // 获取返回值方式二： T get(long timeout, TimeUnit unit) 设置等待超时时间
        Double r2 = supplyFuture.get(5, TimeUnit.SECONDS);
        // 获取返回值方式三：T getNow(T valueIfAbsent) 当有返回结果时则返回结果，如果异步线程抛出异常则返回设置的默认值
        Double r3 = supplyFuture.getNow(10.0);

        /**
         * runAsync用于创建执行无返回值的任务
         */
        CompletableFuture.runAsync(() -> System.out.println("hello,world."));
        CompletableFuture.runAsync(() -> System.out.println("hello,world"), executor);
        /*===================================================================================/


        /*===================================================================================/
        /**
         * 功能:当前任务正常完成以后执行,当前任务的执行结果可以作为下一任务的输入参数,无返回值.
         * 场景:执行任务A,同时异步执行任务B,待任务B正常返回之后,用B的返回值执行任务C,任务C无返回值
         */
        supplyFuture.thenAccept((t) -> System.out.println("price: " + t));
        supplyFuture.thenAcceptAsync((t) -> System.out.println(t));
        supplyFuture.thenAcceptAsync((t) -> System.out.println(t), executor);
        /**
         * 功能:对不关心上一步的计算结果，执行下一个操作
         * 场景:执行任务A,任务A执行完以后,执行任务B,任务B不接受任务A的返回值(不管A有没有返回值),也无返回值
         */
        supplyExecutorFuture.thenRun(() -> System.out.println("thenRun"));
        supplyExecutorFuture.thenRunAsync(() -> System.out.println("thenRunAsync"));
        supplyExecutorFuture.thenRunAsync(() -> System.out.println("thenRunAsyncWithExecutor"), executor);

        /**
         * 功能：当前任务正常完成以后执行，当前任务的执行结果会作为下一任务的执行参数，有返回值
         * 场景：多个任务的串联执行，下一任务的执行依赖于上一任务的结果，每个任务都有输入、输出
         * 实例：异步执行任务A,当任务A完成时使用A的返回结果resultA作为入参进行任务B的处理,可实现任意多个任务的串联执行
         */
        CompletableFuture<Double> applyFuture = supplyFuture.thenApply((s) -> s + 1);
        CompletableFuture<Double> applyFutureAsync = supplyFuture.thenApplyAsync((s) -> s + 1);
        CompletableFuture<Double> applyFutureAsyncExecutor = supplyFuture.thenApplyAsync((s) -> s + 1, executor);


        /**
         * 功能:结合两个CompletionStage的结果，进行转化后返回
         * 场景:需要根据商品id查询商品的当前价格,分两步,查询商品的原始价格和折扣,这两个查询相互独立,
         * 当都查出来的时候用原始价格乘折扣,算出当前价格. 使用方法:thenCombine(..)
         * 如果不需要返回新值就使用 thenAcceptBoth(..)
         * 如果两个任务的返回值也不关心就使用 runAfterBoth()
         */
        CompletableFuture<Double> futurePrice = CompletableFuture.supplyAsync(() -> 100d);
        CompletableFuture<Double> futureDiscount = CompletableFuture.supplyAsync(() -> 0.8);
        CompletableFuture<Double> futureResult = futurePrice.thenCombine(futureDiscount, (price, discount) -> price * discount);
        System.out.println("最终价格为:" + futureResult.join());

        /**
         * 功能:这个方法接收的输入是当前的CompletableFuture的计算值，返回结果将是一个新的CompletableFuture
         * thenApply():它的功能相当于将CompletableFuture<T>转换成CompletableFuture<U>,改变的是同一个CompletableFuture中的泛型类型
         * thenCompose():用来连接两个CompletableFuture，返回值是一个新的CompletableFuture
         */
        CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> "hello");
        CompletableFuture<String> futureB = futureA.thenCompose(s -> CompletableFuture.supplyAsync(() -> s + "world"));
        CompletableFuture<String> futureC = futureB.thenCompose(s -> CompletableFuture.supplyAsync(s::toUpperCase));
        System.out.println(futureC.join());

        /**
         * 功能:执行两个CompletionStage的结果,那个先执行完了,就是用哪个的返回值进行下一步操作
         * 场景:假设查询商品a,有两种方式,A和B,但是A和B的执行速度不一样,我们希望哪个先返回就用那个的返回值
         * acceptEither()
         * runAfterEither()
         */
        CompletableFuture<String> result1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "res1";
        });
        CompletableFuture<String> result2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "res2";
        });
        CompletableFuture<String> result3 = result1.applyToEither(result2, t -> "result: " + t);
        System.out.println(result3.join());

        /**
         * 功能:当运行出现异常时,调用该方法可进行一些补偿操作,如设置默认值.
         * 场景:异步执行任务A获取结果,如果任务A执行过程中抛出异常,则使用默认值100返回.
         */
        CompletableFuture<String> exceptionally = CompletableFuture.supplyAsync(() -> "执行结果：" + (100 / 0))
                .thenApply(s -> "result:" + s)
                .exceptionally(e -> {
                    System.out.println(e.getMessage());
                    return "100";
                });
        System.out.println(exceptionally.join());

        /**
         * 执行流程 supplyAsync->whenComplete->exceptionally
         * 如果 whenComplete与exceptionally互换位置，则在whenComplete中拿到的是exceptionally包装后的值
         */
        CompletableFuture<String> whenComplete = CompletableFuture.supplyAsync(() -> "执行结果：" + (100 / 0))
                .thenApply(s -> "result:" + s)
                .whenComplete((s, t) -> {
                    if (s == null) {
                        System.out.println(s);
                    }
                    if (t == null) {
                        System.out.println(t);
                    }
                })
                .exceptionally(e -> {
                    System.out.println(e.getMessage());
                    return "100";
                });
        System.out.println(whenComplete.join());

        /**
         * 功能:当CompletableFuture的计算结果完成，或者抛出异常的时候，可以通过handle方法对结果进行处理
         */
        CompletableFuture<String> handle = CompletableFuture.supplyAsync(() -> "执行结果：" + (100 / 0))
                .thenApply(s -> "result:" + s)
                .whenComplete((s, t) -> {
                    if (s == null) {
                        System.out.println(s);
                    }
                    if (t == null) {
                        System.out.println(t);
                    }
                })
                .exceptionally(e -> {
                    System.out.println(e.getMessage());
                    return "100";
                })
                .handle((k, v) -> {
                    if (k != null) {
                        System.out.println(k);
                    }
                    if (v != null) {
                        System.out.println(v);
                    }
                    return k + v;
                });


        Thread.sleep(200);
    }

    static Double fetchPrice() {
        try {
            Thread.sleep(100);
        } catch (Exception e) {
            e.printStackTrace();
        }
        if (Math.random() < 0.3) {
            throw new RuntimeException("fetch price failed");
        }

        return 5 + Math.random() * 20;
    }
}
